package com.louis.ylog.appender;

import ch.qos.logback.classic.spi.ILoggingEvent;
import com.dianping.cat.Cat;
import com.louis.ylog.TraceContext;
import com.louis.ylog.feign.CatConstantsExt;
import com.louis.ylog.starter.YLogProperties;
import net.logstash.logback.appender.LogstashTcpSocketAppender;
import net.logstash.logback.encoder.LogstashEncoder;

import java.util.concurrent.atomic.AtomicLong;

public class ELKAppender extends LogstashTcpSocketAppender {

    private final AtomicLong consecutiveDroppedCount;

    private final   boolean enableElk;

    private final boolean enableCat;

    private ELKAppender(){
        this(null);
    }

    public boolean isEnableElk(){
        return this.enableElk;
    }

    public boolean isEnableCat(){
        return this.enableCat;
    }

    public ELKAppender(YLogProperties logProperties){
        this.enableCat=logProperties!=null&&logProperties.getCat()!=null&&logProperties.getCat().isEnableCat();
        this.enableElk=logProperties!=null&&logProperties.getElk()!=null&&logProperties.getElk().isEnableElk();
        this.consecutiveDroppedCount = new AtomicLong();

        if(this.enableElk){
            this.addDestination(logProperties.getElk().getLogstashHost());
            this.setEncoder(new LogstashEncoder());
        }
    }

    @Override
    public void start(){
            super.start();
    }

    @Override
    protected void prepareForDeferredProcessing(final ILoggingEvent event) {
        super.prepareForDeferredProcessing(event);
    }

    @Override
    protected void append(ILoggingEvent event) {
        if(this.enableElk){
            //确保本服务的基础信息都已经初始化
            TraceContext.confirmInitialed();
            syncAppendELK(event);
        }
    }

    protected void syncAppendELK(ILoggingEvent event){
        this.prepareForDeferredProcessing(event);
        long consecutiveDropped;
        if (!this.getDisruptor().getRingBuffer().tryPublishEvent(this.getEventTranslator(), event)) {
            consecutiveDropped = this.consecutiveDroppedCount.incrementAndGet();
            if (consecutiveDropped % (long)this.getDroppedWarnFrequency() == 1L) {
                Cat.logEvent(CatConstantsExt.Type_Log_Appender_Logstash_WarnFrequency,event.getLoggerName(), "-1",consecutiveDropped+"");
                this.addWarn("Dropped " + consecutiveDropped + " events (and counting...) due to ring buffer at max capacity [" + this.getRingBufferSize() + "]");
            }
        } else {
            consecutiveDropped = this.consecutiveDroppedCount.get();
            if (consecutiveDropped != 0L && this.consecutiveDroppedCount.compareAndSet(consecutiveDropped, 0L)) {
                Cat.logEvent(CatConstantsExt.Type_Log_Appender_Logstash_WarnFrequency,event.getLoggerName(), "-1",consecutiveDropped+"");
                this.addWarn("Dropped " + consecutiveDropped + " total events due to ring buffer at max capacity [" + this.getRingBufferSize() + "]");
            }
        }
    }

}
